package com.asiainfo.iam.config;


import com.asiainfo.iam.constant.MQConstant;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PreDestroy;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

@Configuration
// TODO 1
@ConditionalOnProperty(name = "enable.syn.controller", havingValue = "true")
public class RocketMQConfig {

    // 日志
    private  static final Logger log= LoggerFactory.getLogger(RocketMQConfig.class);

    @Value("${rocketmq.namesrv.address}")
    private String namesrvAddr;
    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    @Value("${rocketmq.producer.retry.max-attempts}")
    private int maxRetryAttempts;//最大重试次数

    @Value("${rocketmq.producer.retry.initial-backoff-ms}")
    private int initialBackoffMs;//初始退避时间（毫秒）

    // 保存 Spring 管理的生产者实例
    private DefaultMQProducer producer;



    /**
     * 创建消息生产者
     * @return DefaultMQProducer
     * @throws Exception
     */

    @Bean(destroyMethod ="shutdown") //让Spring自动管理关闭
    public DefaultMQProducer defaultMQProducer() throws MQClientException {
        try {
            // 创建一个消息生产者，并设置一个消息生产者组
            producer = new DefaultMQProducer(producerGroup);
            // 指定Nameserver地址
            producer.setNamesrvAddr(namesrvAddr);
            // 设置更健壮的参数
            producer.setRetryTimesWhenSendFailed(3);            // 发送失败重试次数
            producer.setSendMsgTimeout(3000);                   // 发送消息超时时间
            producer.setRetryAnotherBrokerWhenNotStoreOK(true); // 存储失败时尝试其他Broker

            producer.start();
            log.info("RocketMQ 生产者创建成功！");

            return producer;
        } catch (MQClientException e) {
            log.error("创建消息生产者失败！", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 发送全量同步消息（带重试机制）
     */
    public SendResult sendFullMessage(String tag, String mesBody) throws Exception {
        return sendWithRetry(tag, mesBody,"全量");
    }

    /**
     * 发送增量同步消息（带重试机制）
     */
    public SendResult sendIncrementalMessage(String tag, String mesBody) throws Exception {
        return sendWithRetry(tag, mesBody,"增量");
    }

    /**
     * 发送消息，并添加重试机制
     * @param tag
     * @param mesBody
     * @return
     * @throws Exception
     */
    public SendResult sendWithRetry(String tag, String mesBody,String dataType){
        if(producer==null){
            throw new IllegalStateException(dataType + "数据生产者未初始化！");
        }
        Message message = new Message(
                MQConstant.DATA_SYNC_TOPIC, // 使用同一Topic，保证在同一队列
                tag, // Tag不同，则各Tag会分别被各Consumer消费
                mesBody.getBytes(StandardCharsets.UTF_8) // Body
        );
        int retryCount = 0;//重试次数
        Exception lastException = null;//最后一次异常
        while (retryCount <= maxRetryAttempts) {
            try {
                // 第1次发送（retryCount=0）或重试（retryCount>=1）
                SendResult result = producer.send( message);
                log.debug("{}消息发送成功，tag: {}, msgId: {}", dataType, tag, result.getMsgId());
                return result;
            }catch (Exception e){
                lastException = e;
                // 判断是否需要重试
                if (!isRetryable(e) || retryCount >= maxRetryAttempts) {
                    break;  // 不可重试或达到最大次数，退出循环
                }

                // 计算退避时长 （指数增长）
                long backoffTime = (long) (initialBackoffMs * Math.pow(2, retryCount));
                retryCount++;//重试次数加1
                log.warn("{}消息发送失败，准备第{}次重试（延迟{}ms），原因：{}",
                        dataType,retryCount,backoffTime,e.getMessage());

                try {
                    TimeUnit.MILLISECONDS.sleep(backoffTime);//休眠
                }catch (InterruptedException ie){
                    Thread.currentThread().interrupt(); //保留中断状态
                    throw new RuntimeException("重试等待被中断",ie);
                }
            }
        }
        log.error("{}消息发送失败，已经达到最大重试次数（{}）",
                dataType,maxRetryAttempts,lastException);
        throw new RuntimeException(dataType + "消息发送失败",lastException);

    }

    /**
     *判断异常是否可重试
     */
    private boolean isRetryable(Exception e) {
        // 判断异常类型，是否需要重试
        // 例如，判断是否为网络异常、超时异常、服务端异常等
        // 网络相关异常可重试
        if (e instanceof MQClientException){
            String msg = e.getMessage();
            return msg.contains("connect failed")
                    || msg.contains("timeout")
                    || msg.contains("broker not available");
        }
        // IO异常可重试
        return e instanceof java.io.IOException;

    }


    /**
     * 应用关闭时，关闭 RocketMQ 生产者，释放资源
     */
    @PreDestroy
    public void destroy() {
        if (producer != null) {
            try {
                // 优雅关闭：先将生产者标记为不接收新请求
                producer.shutdown();
                log.info("RocketMQ 生产者已优雅关闭");
            } catch (Exception e) {
                log.error("关闭 RocketMQ 生产者时发生异常", e);
            }
        }
    }

}
